package cn.gwm.flink.streaming.ods;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.function.apply.OdsCoProcessFunction;
import cn.gwm.flink.streaming.function.map.OriginalDataMapFunction;
import cn.gwm.flink.streaming.ods.model.ModelCanEnum;
import cn.gwm.flink.streaming.ods.model.ModelGpsEnum;
import cn.gwm.flink.streaming.ods.model.StandardFieldConf;
import cn.gwm.flink.streaming.sink.hbase.HbaseSinkClient;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.flink.streaming.task.BaseTask;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.db.Db;
import cn.hutool.db.ds.DSFactory;
import cn.hutool.json.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;

/**
 * @ClassName ModelHandle
 * @Author xzh
 * @Date 2022/11/25 16:20
 **/
public class ModelHandle  implements Serializable {

    private static final Logger logger = LoggerFactory.getLogger(ModelHandle.class);
    //车型具体can枚举
    private ModelCanEnum canEnum;
    //车型具体gps枚举
    private ModelGpsEnum gpsEnum;
    //10s信号字段最小长度
    private int minLen;
    //整车类型
    private String vehicleType;
    //can信号保存文件名
    private String canPrefix;
    //gps信号保存文件名
    private String gpsPrefix;
    //3月7、8号添加is_gps_flag字段后路径
    private final String HIVE_PATH = "ods/gps/";
    //信号保存路径
    private final String HIVE_PATH_ORC_G = "ods/";
    //3月7号前保存路径
    public final static String HIVE_PATH_ORC = "odsOrcTxt/";
    public final static String BIZ_TYPE_ODS = "ods";
    //ev纯电  phev混动
    private String powerType;
    //解析信号映射配置文件
    private BeanSource.SourceEnum sourceEnum;
    //合并信号后对应kafka 主题
    private String topic;
    //hbase 表明
    private String hbaseTable;
    //ods层接入字段集合
    List<String> set;

    /**
     *@Author xzh
     *@Description 构成车型处理类
     *@Date 2023/3/3 13:23
     *@Param [canEnum, gpsEnum, minLen, vehicleType, canPrefix, gpsPrefix, powerType, topic]
     *@return
     **/
    public ModelHandle(ModelCanEnum canEnum, ModelGpsEnum gpsEnum, int minLen, String vehicleType, String canPrefix,
                       String gpsPrefix, String powerType ) {
        this.canEnum = canEnum;
        this.gpsEnum = gpsEnum;
        this.minLen = minLen;
        this.vehicleType = vehicleType;
        this.canPrefix = canPrefix;
        this.gpsPrefix = gpsPrefix;
        this.powerType = powerType;
        if(VehicleType.POWER_TYPE_EV.equals(this.powerType)){
            sourceEnum = BeanSource.SourceEnum.evOdsGps;
        }else if(VehicleType.POWER_TYPE_PHEV.equals(this.powerType)){
            sourceEnum = BeanSource.SourceEnum.phevOdsGps;
        }else{
            logger.error("Undefined powerType error {}",powerType);
        }
        initSet();
    }
    /**
     *@Author xzh
     *@Description 构成车型处理类
     *@Date 2023/3/3 13:23
     *@Param [canEnum, gpsEnum, minLen, vehicleType, canPrefix, gpsPrefix, powerType, topic]
     *@return
     **/
    public ModelHandle(ModelCanEnum canEnum, ModelGpsEnum gpsEnum, int minLen, String vehicleType, String canPrefix,
                       String gpsPrefix, String powerType, String topic ) {
        this.canEnum = canEnum;
        this.gpsEnum = gpsEnum;
        this.minLen = minLen;
        this.vehicleType = vehicleType;
        this.canPrefix = canPrefix;
        this.gpsPrefix = gpsPrefix;
        this.powerType = powerType;
        if(VehicleType.POWER_TYPE_EV.equals(this.powerType)){
            sourceEnum = BeanSource.SourceEnum.evOdsGps;
        }else if(VehicleType.POWER_TYPE_PHEV.equals(this.powerType)){
            sourceEnum = BeanSource.SourceEnum.phevOdsGps;
        }else{
            logger.error("Undefined powerType error {}",powerType);
        }
        this.topic = topic;
    }

    /**
     *@Author xzh
     *@Description 构成车型处理类
     *@Date 2023/3/3 13:23
     *@Param [canEnum, gpsEnum, minLen, vehicleType, canPrefix, gpsPrefix, powerType, topic]
     *@return
     **/
    public ModelHandle(ModelCanEnum canEnum, ModelGpsEnum gpsEnum, int minLen, String vehicleType, String canPrefix,
                       String gpsPrefix, String powerType, String topic,String hbaseTable) {
        this.canEnum = canEnum;
        this.gpsEnum = gpsEnum;
        this.minLen = minLen;
        this.vehicleType = vehicleType;
        this.canPrefix = canPrefix;
        this.gpsPrefix = gpsPrefix;
        this.powerType = powerType;
        if(VehicleType.POWER_TYPE_EV.equals(this.powerType)){
            sourceEnum = BeanSource.SourceEnum.evOdsGps;
        }else if(VehicleType.POWER_TYPE_PHEV.equals(this.powerType)){
            sourceEnum = BeanSource.SourceEnum.phevOdsGps;
        }else{
            logger.error("Undefined powerType error {}",powerType);
        }
        this.topic = topic;
        this.hbaseTable = hbaseTable;
        initSet();
    }

    /**
     *初始化 set 集合
     */
    private void initSet() {
        this.set = new ArrayList<>();
        try {
            Db.use(DSFactory.get(ConfigLoader.get("mysql.group.db.config")))
                    .find(StandardFieldConf.getEntityWhere(), StandardFieldConf.class).stream()
                    .filter(data -> BIZ_TYPE_ODS.equals(data.getBizType())
                            &&Integer.parseInt(data.getState())>0
                            &&this.vehicleType.equals(data.getVehicletype()))
                    .sorted(new Comparator<StandardFieldConf>() {
                        @Override
                        public int compare(StandardFieldConf o1, StandardFieldConf o2) {
                            return Integer.parseInt(o1.getId())-Integer.parseInt(o2.getId());
                        }
                    })
                    .forEach(data -> set.add(data.getSemaphore()));
            logger.info("读取mysql 表standard_field_conf 成功");
        } catch (SQLException e) {
            logger.error("读取mysql 表standard_field_conf 失败");
            throw new RuntimeException(e);
        }
    }
    public List<String> getSet(){
        return set;
    }

    /**
     *@Author xzh
     *@Description 获取具体车型对应的侧输出流
     *@Date 2023/3/3 13:19
     *@Param [canProcess, canTagMap] 源数据流，侧输出流分类
     *@return org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator<cn.hutool.json.JSONObject>
     **/
    public SingleOutputStreamOperator<JSONObject> getCanStream(SingleOutputStreamOperator<String> canProcess,
                                                               Map<String, OutputTag<String>> canTagMap){
        return getCanStream(canProcess,canTagMap,0);
    }
    public SingleOutputStreamOperator<JSONObject> getCanStream(SingleOutputStreamOperator<String> canProcess,
                                                               Map<String, OutputTag<String>> canTagMap, int minLen){
        return canProcess.getSideOutput(canTagMap.get(canEnum.getModelName()))
                .flatMap(new OriginalDataMapFunction(vehicleType,minLen))
                .returns(JSONObject.class).name("get"+minLen+canEnum.getValue());
    }

    /**
     *@Author xzh
     *@Description 获取具体车型对应的侧输出流
     *@Date 2023/3/3 13:19
     *@Param [canProcess, canTagMap] 源数据流，侧输出流分类
     *@return org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator<cn.hutool.json.JSONObject>
     **/
    public SingleOutputStreamOperator<JSONObject> getGpsStream(SingleOutputStreamOperator<String> gpsProcess, Map<String, OutputTag<String>> tagMap){
        SingleOutputStreamOperator<JSONObject> process = gpsProcess.getSideOutput(tagMap.get(gpsEnum.getModelName()))
                .flatMap(new OriginalDataMapFunction(vehicleType,0)).returns(JSONObject.class).name("get"+gpsEnum.getValue());
        return process;
    }
    /**
     *@Author xzh
     *@Description gps数据以orc格式，按字段配置文件写入hive
     *@Date 2022/12/21 10:47
     *@Param [gpsProcess] 具体车型数据流
     *@return void
     **/
    public void addGpsToHive(DataStream<JSONObject> process){
        //3月7号gps数据增加is_gps_flag 内网
        //3月7号gps数据增加is_gps_flag
        process.addSink(BaseTask.getOrcTxtSink(gpsPrefix,HIVE_PATH, BeanSource.SourceEnum.gpsIntranetFlag)).name("toHive"+vehicleType);
    }


    /**
     *@Author xzh
     *@Description can数据写入默认kafka主题
     *@Date 2022/12/21 10:48
     *@Param [canStream] 具体车型can数据流
     *@return void
     **/
    public void addCanToDefaultKafka(DataStream<JSONObject> canStream){
        canStream.addSink(FlinkKafkaUtil.toKafka(canEnum.getTopic())).name("toKafkaDefault"+vehicleType);
    }

    /**
     *@Author xzh
     *@Description can信号，gps信号 流合并，输出合并后的流
     *@Date 2023/3/3 13:24
     *@Param [canProcess, gpsProcess, canTagMap, gpsTagMap] can源数据流，gps源数据流，输出流分类，输出流分类
     *@return org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator<cn.hutool.json.JSONObject>
     **/
    public SingleOutputStreamOperator<JSONObject> connectStream(SingleOutputStreamOperator<String> canProcess,SingleOutputStreamOperator<String> gpsProcess,
                                                                Map<String, OutputTag<String>> canTagMap, Map<String, OutputTag<String>> gpsTagMap){
        SingleOutputStreamOperator<JSONObject> canStream = canProcess.getSideOutput(canTagMap.get(canEnum.getModelName()))
                .flatMap(new OriginalDataMapFunction(vehicleType,minLen,sourceEnum,set)).returns(JSONObject.class)
                .name(canEnum.getValue()).keyBy(item -> item.getStr(BaseFields.vin))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> element.getLong(BaseFields.tid))).name("canWatermarks"+vehicleType);
        SingleOutputStreamOperator<JSONObject> gpsStream = gpsProcess.getSideOutput(gpsTagMap.get(gpsEnum.getModelName()))
                .flatMap(new OriginalDataMapFunction(vehicleType,0))
                .returns(JSONObject.class).name(gpsEnum.getValue()).keyBy(item -> item.getStr(BaseFields.vin))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner((element, recordTimestamp) -> element.getLong(BaseFields.tid))).name("gpsWatermarks"+vehicleType);
        SingleOutputStreamOperator<JSONObject> process = canStream.keyBy(data -> data.getStr(BaseFields.vin)).connect(gpsStream.keyBy(data -> data.getStr(BaseFields.vin)))
                .keyBy(data -> data.getStr(BaseFields.vin), data -> data.getStr(BaseFields.vin))
                .process(new OdsCoProcessFunction()).name("coProcess"+vehicleType);
        process.addSink(FlinkKafkaUtil.toKafka(topic)).name("toKafka"+vehicleType);
        return process;
    }


    /**
     *@Author xzh
     *@Description can数据流sink到hive
     *@Date 2023/3/3 13:26
     *@Param [process]
     *@return void
     **/
    public void addCanToHive(DataStream<JSONObject> process){
        process.addSink(BaseTask.getOrcTxtSink(canPrefix,HIVE_PATH_ORC_G, sourceEnum)).name("toHive"+vehicleType);
    }

    /**
     *@Author xzh
     *@Description  gps信号流输出到hbase
     *@Date 2023/3/3 13:27
     *@Param [process]
     *@return void
     **/
    public void addGpsToHbase(DataStream<JSONObject> process){
        HbaseSinkClient.hbaseSink(process, hbaseTable,
                ConfigLoader.get("hbase.table.columnfamily"),"toHbase"+vehicleType);
    }
}
